package org.eclipse.moquette.server.netty;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.eclipse.moquette.proto.Utils;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.proto.messages.PingRespMessage;
import org.eclipse.moquette.spi.IMessaging;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.eclipse.moquette.proto.messages.AbstractMessage.*;

import java.util.HashMap;
import java.util.Map;

/**
 * 处理连接MQTT 连接, 除了PingResp 其他的command 都委托给SimpleMessage
 * Created by alex on 15-5-30.
 */
@ChannelHandler.Sharable//注解@Sharable能够让它在channels间共享
public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {

    private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);

    private IMessaging m_messaging;
    //这里存放NettyChannel 和 ctx, TODO 不知道这样存储会有性能问题, 为什么没有并发库
    private final Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<>();

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
        AbstractMessage msg = (AbstractMessage) message;
        LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));

        try {
            switch (msg.getMessageType()) {
                case CONNECT:
                case SUBSCRIBE:
                case UNSUBSCRIBE:
                case PUBLISH:
                case PUBREC:
                case PUBCOMP:
                case PUBREL:
                case DISCONNECT:
                case PUBACK:
                    NettyChannel channel;
                    synchronized (m_channelMapper) {
                        if (!m_channelMapper.containsKey(ctx)) {
                            m_channelMapper.put(ctx, new NettyChannel(ctx));
                        }
                        channel = m_channelMapper.get(ctx);
                    }

                    m_messaging.handleProtocolMessage(channel, msg);
                    break;
                case PINGREQ:
                    PingRespMessage pingResp = new PingRespMessage();
                    ctx.writeAndFlush(pingResp);
                    break;
            }
        } catch (Exception ex) {
            LOG.error("Bad error in processing the message", ex);
        }
    }

    //关闭Channel 和 清除相应的Entry
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        NettyChannel channel = m_channelMapper.get(ctx);
        String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
        m_messaging.lostConnection(channel, clientID); //得到clientID 处理连接断开的事件
        ctx.close();

        synchronized (m_channelMapper) {
            m_channelMapper.remove(ctx);
        }
    }

    public void setMessaging(IMessaging messaging) {
        m_messaging = messaging;
    }
}
